package com.beifeng.mapreduce;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class MyMapReduce extends Configured implements  Tool {
	//KEYIN, VALUEIN, KEYOUT, VALUEOUT 传入的Key和Value类型，传出去的Key和Value类型
	static class MyMap extends Mapper<LongWritable, Text, Text, IntWritable> {
		Text text = new Text();
		IntWritable mr_value = new IntWritable(1);
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
		 String line = 	value.toString();
		 System.out.println("\n=======每一个map输入的偏移量：\n"+key.get()+"\n=============\n");
		 String[] words = line.split(" ");
		 for (String word : words) {
			 text.set(word);
			 context.write(text, mr_value);
		}
		 
		
		}
		
	}
	//KEYIN, VALUEIN, KEYOUT, VALUEOUT   输入keyvalue，输出keyvalue的类型
	static class MyReduce extends Reducer<Text, IntWritable,Text,IntWritable>{
		 private IntWritable result = new IntWritable();
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			int sum = 0;
			for (IntWritable value : values) {
				sum+=value.get();
			}
			result.set(sum);
			context.write(key, result);
		}
		
	}
	// optimize by combiner
	static class Mycombiner extends Reducer<Text, IntWritable,Text,IntWritable>{
		 private IntWritable result = new IntWritable();
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			System.out.println("=============");
			System.out.println("enter combiner phase");
			int sum = 0;
			for (IntWritable value : values) {
				sum+=value.get();
			}
			System.out.println("key : "+key.toString()+"------> "+sum);
			result.set(sum);
			context.write(key, result);
			System.out.println("combiner end");
		}
		
	}
	


	public int run(String[] args) throws Exception {
		// TODO Auto-generated method stub
		//hadoop的八股文
		//Configuration conf = new Configuration();
		Configuration conf = this.getConf();
		Job job = Job.getInstance(conf, this.getClass().getSimpleName());
		//对job进行具体的配置
		
		//当你本地运行，这个设置可以不写，不会报错
		//当提价到集群上面运行的时候，这个设置不写，会报类找不到的异常 
		job.setJarByClass(MyMapReduce.class);
		
		//写一个输入路径
		Path input = new Path(args[0]);
		FileInputFormat.addInputPath(job, input);
		//写一个输出路径
		Path output = new Path(args[1]);
		//if the output folder exists , delete it start
		FileSystem fileSystem =output.getFileSystem(conf);
		FileOutputFormat.setOutputPath(job, output);
		if(fileSystem.exists(output)){
			fileSystem.delete(output, true);
		}
		//if the output folder exists , delete it end
		//==============shuffle=======================
		job.setCombinerClass(Mycombiner.class);
		//==============shuffle=======================
		//设置运行的map类的相关参数
		job.setMapperClass(MyMap.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//设置运行的Reduce的相关参数
		job.setReducerClass(MyReduce.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		boolean isSuccess =  job.waitForCompletion(true);
		
		
		return isSuccess?0:1;
	}
	
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		args = new String[]{
				"hdfs://centos6-1:8020/data.txt",
				"hdfs://centos6-1:8020/wc/output5"
		};
		/*MyMapReduce test =  new MyMapReduce();
		try {
			test.run(args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
		Configuration conf = new Configuration();
		try {
			ToolRunner.run(conf,new MyMapReduce(), args);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
 	}

}
